package com.examples.mysql2mysql.binlog;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDCMain {

    /** 该方法不可用 **/
    public static void main(String[] args) throws Exception {
        //用于读取mysql-binlog数据配置  connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/f_saas_cfg?useUnicode=true&characterEncoding=UTF-8&useAffectedRows=true&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8", "root", "123456");
        MySqlSourceBuilder<String> mySqlSourceBuilder = MySqlSource.builder();
        MySqlSource<String> mySqlSource = mySqlSourceBuilder
                .hostname("127.0.0.1")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("f_saas_cfg")
                //.tableList("f_saas_cfg.test_table","f_saas_cfg.test_table01")
                .tableList("f_saas_cfg.test_table")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .includeSchemaChanges(true)
                .build();
        System.out.println("init data source");


        Configuration configuration=new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.enableCheckpointing(10);
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MYSQL Source").addSink(new CustomSink());
        env.execute();

        System.out.println("env start");

    }


}
